package com.atguigu.flink.state.keyedstate;

import com.atguigu.flink.func.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by 黄凯 on 2023/6/20 0020 19:58
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 检测每种传感器的水位值，如果连续的两个水位值超过10，就输出报警
 */
public class Flink05_ValueState {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(2000);

        SingleOutputStreamOperator<WaterSensor> ds = env
                .socketTextStream("127.0.0.1", 8888)
                .map(new WaterSensorMapFunction());

        KeyedStream<WaterSensor, String> keyedDS = ds.keyBy(WaterSensor::getId);

        //使用keyedStream之后在处理时，使用状态，全部都是 KeyedState
        keyedDS
                .process(
                        new KeyedProcessFunction<String, WaterSensor, String>() {

                            /**
                             * 把当前传感器的水位保存下来
                             * 写：
                             * lastVc.update(T ):更新存储的单个值
                             * 读：
                             * lastVc.value():读取存储的单个值
                             *
                             */
                            private ValueState<Integer> lastVc;

                            /**
                             * 初始化方法
                             */
                            @Override
                            public void open(Configuration parameters) throws Exception {

                                //找到状态的备份存储，取出状态并赋值
                                lastVc = getRuntimeContext().getState(new ValueStateDescriptor<>("lastVc", Integer.class));


                            }

                            /**
                             * KeyedProcessFunction<K, I, O> extends AbstractRichFunction implements RichFunction, Serializable
                             *
                             * 生命周期方法：
                             *      open():在task被创建后，执行，就是初始化方法
                             *      close()
                             */
                            @Override
                            public void processElement(WaterSensor waterSensor,
                                                       KeyedProcessFunction<String, WaterSensor, String>.Context context,
                                                       Collector<String> collector) throws Exception {

                                if ("x".equals(waterSensor.getId())){
                                    throw new RuntimeException("出异常了！！！");
                                }

                                //先获取上一个WaterSensor的水位
                                Integer lastVcNum = lastVc.value();

                                //如果连续的两个水位值超过10，就输出报警
                                if (lastVcNum != null && lastVcNum > 10 && waterSensor.getVc() > 10) {

                                    collector.collect("连续的两个水位值超过10,报警!");

                                } else {

                                    collector.collect(waterSensor.toString());

                                }

                                //把当前传感器的水位，存入状态
                                lastVc.update(waterSensor.getVc());


                            }
                        }
                ).print();

        env.execute();

    }

}
